package com.shujia.useraction

import com.shujia.utils.{Constants, SparkMain}
import org.apache.spark.sql.{DataFrame, SaveMode}

class UserActionSkill(day: String) extends UserActionBase {

  import spark.implicits._
  import org.apache.spark.sql.functions._


  override def load(path: String, day: String): DataFrame = {
    spark
      .read
      .format("csv")
      .option("sep", "#")
      .schema("date String,cid String,stbid String,utctime String,semanticType String,text String,intent String,t String,semanticStr String")
      .load(s"$path/$day")

  }

  override def process(): Unit = {
    df = load(Constants.USER_ACTION_BASE_SKILL_PATH, day)
      .where($"intent" === "query")
      .withColumn("uid", upper(md5(concat($"stbid", $"cid"))))
      .select($"uid", $"text" as "itemName", $"utctime")
  }

  override def save(): Unit = {
    df
      .write
      .format(Constants.USER_ACTION_SKILL_FORMAT)
      .mode(SaveMode.Overwrite)
      .save(s"${Constants.USER_ACTION_SKILL_PATH}/$day")
  }

  override def load(): DataFrame = {
    spark
      .read
      .format(Constants.USER_ACTION_SKILL_FORMAT)
      .load(s"${Constants.USER_ACTION_SKILL_PATH}/$day")
  }
}

object UserActionSkill extends SparkMain {
  override def run(day: String): Unit = {

    val userActionSkill: UserActionSkill = UserActionSkill(day)

    userActionSkill.process()
    userActionSkill.save()
  }

  def apply(day: String): UserActionSkill = new UserActionSkill(day)
}
